package com.bw.gmall.realtime.app.dwd;

import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.ZoneId;

public class DwdUserRegister {
    public static void main(String[] args) throws Exception {

        //todo 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        tenv.getConfig().setLocalTimeZone(ZoneId.of("GMT+8"));

        // TODO 3. 从 Kafka 读取业务数据，封装为 Flink SQL 表
        tenv.executeSql("create table topic_db(" +
                "`database` string, " +
                "`table` string, " +
                "`type` string, " +
                "`data` map<string,string>, " +
                "`ts` string " +
                ")"+ MyKafkaUtil.getKafkaDDL("topic_db","dwd_user_register"));
        // TODO 4. 读取用户表数据
        Table userInfo = tenv.sqlQuery("select " +
                "data['id'] user_id, " +
                "data['create_time'] create_time, " +
                "ts " +
                "from topic_db " +
                "where `table` = 'user_info' " +
                "and `type` = 'insert' ");
        tenv.createTemporaryView("user_info",userInfo);
        // TODO 5. 创建 Kafka-Connector dwd_user_register 表
        tenv.executeSql("create table `dwd_user_register` (" +
                "user_id string, " +
                "date_id string, " +
                "create_time string, " +
                "ts string " +
                ") " + MyKafkaUtil.getKafkaSinkDDL("dwd_user_register"));
        // TODO 6. 将输入写入 Kafka-Connector 表
        tenv.executeSql("insert into dwd_user_register select " +
                "user_id, " +
                "date_format(create_time,'yyyy-MM-dd') date_id, " +
                "create_time, " +
                "ts " +
                "from user_info");
        tenv.sqlQuery("select * from dwd_user_register").execute().print();

        env.execute();
    }
}
